package com.atguigu.gmall.realtime.common.base;

import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSourceUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION;

public abstract class BaseApp {
    /**
     * 未来每个具体的App程序需要按照自身所关心的处理过程, 对数据进行处理
     * @param env
     * @param stream
     */
    public abstract void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream);

    /**
     * 通过start方法整体控制Flink程序的流程
     * @param port 每个App程序的监控界面使用的端口
     *              （1）DIM层维度分流应用使用10001端口
     *              （2）DWD层应用程序按照在本文档中出现的先后顺序，端口从10011开始，自增1
     *              （3）DWS层应用程序按照在本文档中出现的先后顺序，端口从10021开始，自增1
     * @param parallelism
     *              默认4个
     * @param ckAndGroupId
     *              Job主程序类名的下划线命名形式, 例如:DimApp ==> dim_app
     * @param topic
     */
    public void start(int port, int parallelism, String ckAndGroupId, String topic){
        // 1.准备执行环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", port);  // conf.setInteger(RestOptions.PORT, port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 设置并行度
        env.setParallelism(parallelism);
        System.setProperty("HADOOP_USER_NAME", Constant.HDFS_USER_NAME);

        // 2.状态后端&检查点相关设置
        // 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        // 开启 checkpoint
        env.enableCheckpointing(5000);
        // 设置 checkpoint 模式: 精准一次
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // checkpoint 存储
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall2023/stream/" + ckAndGroupId);
        // checkpoint 并发数
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // checkpoint 之间的最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
        // checkpoint  的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        // job 取消时 checkpoint 保留策略
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(RETAIN_ON_CANCELLATION);

        // 3. 创建KafkaSource对象, 从Kafka中读取数据
        KafkaSource<String> kafkaSource = FlinkSourceUtil.getKafkaSource(topic, ckAndGroupId);

        DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        // 4.对数据进行转换处理
        handle(env, stream);

        // 5. 执行
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
